package project

import dmputils.PeiZhiFile
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{Row, SQLContext}

import org.apache.spark._

/**
  * Created by Administrator on 2018/03/31.
  */
object DataBiaoQian {
  def main(args: Array[String]): Unit = {


  val conf = new SparkConf()
    .setMaster("local[*]")
    .setAppName("数据标签化")
    .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

  val sc: SparkContext = new SparkContext(conf)
  val sqlc: SQLContext = new SQLContext(sc)
  import sqlc.implicits._

  //广播APP名称词典
  val app_dictfile: RDD[String] = sc.textFile(PeiZhiFile.config.getString("app_dictpath"))
  val splitApp_dictData: RDD[Array[String]] = app_dictfile.map(_.split("\t"))
  val GlApp_dictData: RDD[(String, String)] = splitApp_dictData.map(t=>{
    val tuple = if (t.length >= 5) {
      (t(4),t(1))
    } else {
      (null, null)
    }
    tuple
  })
  val GlApp_dictDatabroadcast: Broadcast[RDD[(String, String)]] = sc.broadcast(GlApp_dictData)
  val GlApp_dictmap: Map[String, String] = GlApp_dictDatabroadcast.value.map(t=>(t._1,t._2)).collect().toMap

  val stopwordsfile: RDD[String] = sc.textFile(PeiZhiFile.config.getString("stopwords_path"))
    val stopWordsbroadcast: Broadcast[RDD[String]] = sc.broadcast(stopwordsfile)

  val dataFrame = sqlc.read.parquet(PeiZhiFile.config.getString("inputpath"))
  val filteredData: RDD[Row] = dataFrame.map(x => x).filter(df =>
           !df.getAs[String]("imei").isEmpty
        || !df.getAs[String]("mac").isEmpty
        || !df.getAs[String]("idfa").isEmpty
        || !df.getAs[String]("openudid").isEmpty
        || !df.getAs[String]("androidid").isEmpty)

    val GLData= filteredData.map(df => {
      val imei = df.getAs[String]("imei")
      val mac = df.getAs[String]("mac")
      val idfa = df.getAs[String]("idfa")
      val openudid = df.getAs[String]("openudid")
      val androidid = df.getAs[String]("androidid")
      //设备类型 （1：android 2：ios 3：wp）
      val client = df.getAs[Int]("client")
      //渠道
      val adplatformproviderid = ("CN" + df.getAs[String]("adplatformproviderid"), 1)
      //省份
      val provincename = ("ZP" + df.getAs[String]("provincename"), 1)
      //城市
      val cityname = ("ZC" + df.getAs[String]("cityname"), 1)
      //广告位类型（1：banner 2：插屏 3：全屏）
      val adspacetype = df.getAs[Int]("adspacetype")
      //广告位类型名称
      val adspacetypename = ("LN" + df.getAs[String]("adspacetypename"), 1)
      //关键字
      val keyword = df.getAs[String]("keywords")
      //APP名称
      val appname = df.getAs[String]("appname")
      //APPID
      val appid = df.getAs[String]("appid")
      //  联网方式名称 1-2G 2-3G 3-4G 4-未知 5-WIFI
      val networkmannername = df.getAs[String]("networkmannername")
      //运营商名称 1-移动 2-联通 3-电信 4-未知
      val ispid = df.getAs[Int]("ispid")
      //获取IMEI值
      val IMEI: String = if (!imei.isEmpty) {
        imei
      } else if (!mac.isEmpty) {
        mac
      } else if (!idfa.isEmpty) {
        idfa
      } else if (!openudid.isEmpty) {
        openudid
      } else {
        androidid
      }
      //获取当前设备类型
      val SBLX: (String, Int) = if (client == 1) {
        ("D00010001", 1)
      } else if (client == 2) {
        ("D00010002", 1)
      } else if (client == 3) {
        ("D00010003", 1)
      } else {
        ("D00010004", 1)
      }
      //转换广告类型
      val adType: (String, Int) = if (adspacetype < 10) {
        ("LC0" + adspacetype, 1)
      } else {
        ("LC" + adspacetype, 1)
      }
      //获取APP名称
      val newappname: (String, Int) = if (appname.equals("其他") || appname.equals("未知")) {
        val appname1: String = GlApp_dictmap.getOrElse(appid, appid).toString
        (appname1, 1)
      } else {
        (appname, 1)
      }
      //联网方式转换
      val newnetworkmannername: (String, Int) = if (networkmannername == "2G") {
        ("D00020004", 1)
      } else if (networkmannername == "3G") {
        ("D00020003", 1)
      } else if (networkmannername == "4G") {
        ("D00020002", 1)
      } else if (networkmannername == "Wifi") {
        ("D00020001", 1)
      } else {
        ("D00020005", 1)
      }
      //运营商名称 1-移动 2-联通 3-电信 4-未知
      val newispid: (String, Int) = if (ispid == 1) {
        ("D00030001", 1)
      } else if (ispid == 2) {
        ("D00030002", 1)
      } else if (ispid == 3) {
        ("D00030003", 1)
      } else {
        ("D00030004", 1)
      }
      val stopWordsvalue: RDD[String] = stopWordsbroadcast.value

      val A: Array[String] = stopWordsvalue.toArray()

      val keywords: Array[String] = keyword.split("[|]")

      val filteredWords: Array[String] = keywords.filter(t => t.length >= 3 && t.length <= 8)
        .filter(x => !x.equals(A(0)) && !x.equals(A(1)) && !x.equals(A(2)) && !x.equals(A(3)))
      val wordAndOne: Array[(String, Int)] =filteredWords.map(t => {
        ("K" + t, 1)
      })
      (IMEI, List(adType,
        adspacetypename,
        newappname,
        adplatformproviderid,
        SBLX,
        newnetworkmannername,
        newispid,
        provincename,
        cityname) ++
        wordAndOne)
      //IMEI 广告类型     广告位名称      APPname        渠道           设备类型      联网方式            运营商      省份         城市
    })


    // RDD[(String, Lis((String, Int), (String, Int), (String, Int), (String, Int), (String, Int), (String, Int), (String, Int), (String, Int), (String, Int), Array[(String, Int)])]













//    val GLDataDf = GLDatafilter.toDF("IMEI","广告类型","广告位名称","APPname","渠道","设备类型","联网方式","运营商","省份","城市","wordAndOne")
//    GLDataDf.registerTempTable("t_biaoqian")
//    sqlc.sql("select * from t_biaoqian" ).show(1000)






  sc.stop()


  }

}
